package com.bestv.bsr.cloud.service.impl;

import com.bestv.bsr.cloud.common.DisplayNotifyService;
import com.bestv.bsr.core.annotations.OptionalComponent;
import com.bestv.bsr.core.kafka.BSRMsg;
import com.bestv.bsr.core.properties.GlobalProperties;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.jetbrains.annotations.NotNull;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

/**
 * @author taojiacheng
 */
@OptionalComponent(value = "displayNotifyService", strategy = OptionalComponent.Strategy.CLOUD)
@Slf4j
@RequiredArgsConstructor
public class CloudDisplayNotifyServiceImpl implements DisplayNotifyService {

    private final GlobalProperties globalProperties;

    /**
     * todo 卡夫卡配置
     * todo traceId 传递
     */
    private final KafkaTemplate<String, BSRMsg<?>> kafkaTemplate;

    @Override
    public <T> void notify(BSRMsg<T> msg) {
        log.info("cloud notify [{}]", msg);
        ListenableFuture<SendResult<String, BSRMsg<?>>> future = kafkaTemplate.send(globalProperties.getNotifyTopic(), msg);
        future.addCallback(new ListenableFutureCallback<SendResult<String, BSRMsg<?>>>() {
            @Override
            public void onFailure(@NotNull Throwable throwable) {
                //发送失败
                log.info("send kafka failure [{}], throwable [{}]", msg, throwable.getMessage());
            }

            @Override
            public void onSuccess(SendResult<String, BSRMsg<?>> stringObjectSendResult) {
                //发送成功
                log.info("send kafka success [{}]", msg);
            }
        });
    }
}
